import kafka.serializer.StringDecoder
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.text.SimpleDateFormat
import java.util.Date

/**
 * 7.3	最近1小时广告点击量实时统计
 *
 * 需求：统计各广告最近1小时内的点击量趋势，每6s更新一次（各广告最近1小时内各分钟的点击量）
 *
 * 1.最近一小时             => 窗口的长度为1小时
 * 2.每6s更新一次           => 窗口的滑动步长是6s
 * 3.各个广告每分钟的点击量   => ((广告id,时间),1)
 */
object Request2 {
  def main(args: Array[String]): Unit = {
    // 创建Spark配置文件对象
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("Request2")
    // 创建SparkStreamingContext对象
    val ssc: StreamingContext = new StreamingContext(conf, Seconds(3))
    // 设置检查点
    ssc.checkpoint("C:\\Users\\22734\\IdeaProjects\\spark-api\\spark-api-realtime\\checkpoint")
    // kafka参数声明
    val brokers: String = "hadoop102:9092,hadoop103:9092,hadoop104:9092"
    val topic: String = "my-ads-bak"
    val group: String = "bigdata"
    val deserialization: String = "org.apache.kafka.common.serialization.StringDeserializer"

    val kafkaParams: Map[String, String] = Map(
      ConsumerConfig.GROUP_ID_CONFIG -> group,
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> brokers,
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> deserialization,
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> deserialization
    )

    // 创建DS
    val kafkaDS: InputDStream[(String, String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, Set(topic))

    // 测试 kafka中消费数据 msg = 1584271384370,华南,广州,100,1
    val dataDS: DStream[String] = kafkaDS.map(_._2)

    // 定义窗口。窗口大小12s，滑动步长3s
    val windowDS: DStream[String] = dataDS.window(Seconds(12), Seconds(3))

    // 转换结构为((广告id,时间), 1)
    val mapDS: DStream[((String, String), Int)] = windowDS.map {
      line => {
        val fields: Array[String] = line.split(",")
        val date: Date = new Date(fields(0).toLong)
        val sdf: SimpleDateFormat = new SimpleDateFormat("hh:mm")
        val dateStr: String = sdf.format(date)
        ((fields(4), dateStr), 1)
      }
    }
    // 对数据进行聚合
    val resDS: DStream[((String, String), Int)] = mapDS.reduceByKey(_ + _)
    resDS.print()

    ssc.start()
    ssc.awaitTermination()
  }
}
